package com.leilei.broadcaststate.common;


import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/24 21:55
 */
public class VehicleOrgRelationSource extends RichSourceFunction<Tuple2<Integer, List<Integer>>> {
    private boolean flag = true;
    private Connection conn = null;
    private PreparedStatement ps = null;
    private ResultSet rs = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        conn = DriverManager.getConnection("jdbc:mysql://xxx:3306/flink-learn", "root", "xx");
        String sql = "select vehicle_id,group_concat(org_id) as orgIdList from relation_organization_vehicle group by vehicle_id;";
        ps = conn.prepareStatement(sql);
    }

    @Override
    public void run(SourceContext<Tuple2<Integer, List<Integer>>> context) throws Exception {
        while (flag) {
            ResultSet rs = ps.executeQuery();
            while (rs.next()) {
                VehicleOrgRelation vehicleOrgRelation = new VehicleOrgRelation(
                        rs.getInt("vehicle_id"),
                        Arrays.stream(rs.getString("orgIdList").split(","))
                                .parallel()
                                .map(Integer::valueOf)
                                .collect(Collectors.toList()));
                context.collect(Tuple2.of(vehicleOrgRelation.getVehicleId(), vehicleOrgRelation.getOrgIdList()));
            }
            Thread.sleep(60000);
        }

    }

    @Override
    public void cancel() {
        flag = false;
    }

    @Override
    public void close() throws Exception {
        if (conn != null) {
            conn.close();
        }
        if (ps != null) {
            ps.close();
        }
        if (rs != null) {
            rs.close();
        }
    }
}
